Snowflakeでドキュメントデータを処理しよう！Document AIのチュートリアル「Tutorial: Create a document processing pipeline with Document AI」をやってみた

さがらです。

先月Document AIがプレビューとなりました。2024年6月29日時点ではAWSの東京リージョンでも使用可能となっています。

このDocument AI機能を一通り試すことが出来るチュートリアルとして「Tutorial: Create a document processing pipeline with Document AI」があります。今回このチュートリアルをやってみたので、その内容をまとめてみます。

Document AIとは

Document AIは、Snowflake独自の大規模言語モデル (LLM) であるArctic-TILT を使用してドキュメントからデータを抽出する機能です。

手書きのテキストだけでなく、ロゴやチェックマークなども読み取り情報を抽出することが可能です。

下記のYouTubeのデモがDocument AIのイメージを掴むにはぴったりだと思います。このデモのようにドキュメントから情報を抽出することに加え、本チュートリアルで行うように構造化データ(テーブル)に変換するパイプラインを構成することも可能です。

Document AIのコストについて

Document AIは通常のウェアハウスとストレージコストに加え、「AI Services compute」の費用が発生します。

下図は公式ドキュメントからの引用ですが、文書の複雑度、文書の数、取得する値の数、に応じて消費するクレジットの概算が書かれていますので、こちらを参考にすると良いと思います。

また、画像引用元の公式ドキュメントも併せてご覧ください。

Document AIに関する注意事項

Document AIは非常に協力な機能ですが、いくつか注意すべきポイントがあります。（2024年6月時点）

• Document AIは英語のドキュメントの処理をサポートしており、他の言語では満足のいく結果が得られない可能性があること
• 関連する権限を直接付与したロールでないとDocument AIは操作ができないこと（例えば、Document AI関連の権限を付与したロールをSYSADMINの配下にして、SYSADMINでDocument AIを操作しようとしても出来ないようになっています）
• PDF、PNG、DOCX、EML、JPEG、JPG、HTM、HTML、TEXT、TXT、TIF、TIFFをサポートしていること
• ドキュメントのは125ページ以内であること
• ドキュメントのサイズは50MB以下であること

対応しているドキュメントの種類と関連する制約については、下記のドキュメントをご覧ください。

他の注意事項については、下記のドキュメントをご覧ください。

Introduction

ここからは実際にDocument AIのチュートリアルに沿ってやってみた内容をまとめてみます。

まずはIntroductionです。

ここでは、本チュートリアルで学ぶことやDocumenti AIの概要や使う上での前提条件の説明がありますね。

下記はWhat you will learnからの引用ですが、本チュートリアルでは下記のことを学べるようになっています。

• Document AIを操作するために必要なオブジェクトと権限を設定します
• SnowsightのDocument AIユーザーインターフェースを使用して Document AI model buildを準備し、非構造化ドキュメントからデータを抽出します
• Document AI model build、ストリーム、タスクを使用して、段階的に新しいドキュメントを継続的に処理するためのパイプラインを作成します

また、本チュートリアルを進める前提条件として以下についても言及されています。

• ACCOUNTADMINロールを操作できるユーザーであること
• 商用アカウントであること
• 日本ではAWS大阪リージョンは対応していないこと（AWS東京、Azure東京はOK）

Set up the required objects and privileges

ここでは、Document AI model buildを格納するデータベースとスキーマ、Document AI model buildを作りドキュメント処理パイプラインを構築するためのカスタムロールの作成を行います。

チュートリアル上のクエリに対して、ウェアハウスの作成やカスタムロールをSYSADMINの配下にすることを追加したクエリが下記となります。

最後のクエリの<your_user_name>だけ対象のユーザーに書き換えて、一通り実行してください。

-- Document AI model buildを作成するためのデータベース・スキーマ・ウェアハウスを作成
USE ROLE SYSADMIN;
CREATE DATABASE doc_ai_db;
CREATE SCHEMA doc_ai_db.doc_ai_schema;

CREATE WAREHOUSE doc_ai_wh
    WAREHOUSE_SIZE = XSMALL
    AUTO_SUSPEND = 60
    AUTO_RESUME = TRUE
    INITIALLY_SUSPENDED = TRUE;

-- Document AI model buildを準備し、処理パイプラインを作成するためのカスタムロールを作成します
USE ROLE ACCOUNTADMIN;

CREATE ROLE doc_ai_role;

-- カスタムロールにSNOWFLAKE.DOCUMENT_INTELLIGENCE_CREATORデータベースロールを付与
GRANT DATABASE ROLE SNOWFLAKE.DOCUMENT_INTELLIGENCE_CREATOR TO ROLE doc_ai_role;

-- カスタムロールにウェアハウスのUSAGE権限とOPERATE権限を付与
GRANT USAGE, OPERATE ON WAREHOUSE doc_ai_wh TO ROLE doc_ai_role;

-- カスタムロールに作成したデータベースとスキーマを使用する権限を付与
GRANT USAGE ON DATABASE doc_ai_db TO ROLE doc_ai_role;
GRANT USAGE ON SCHEMA doc_ai_db.doc_ai_schema TO ROLE doc_ai_role;

-- カスタムロールに抽出用のドキュメントを保存するスキーマに対するステージ作成権限を付与
GRANT CREATE STAGE ON SCHEMA doc_ai_db.doc_ai_schema TO ROLE doc_ai_role;

-- カスタムロールにmodel build (DOCUMENT_INTELLIGENCEクラスのインスタンス) を作成する権限を付与
GRANT CREATE SNOWFLAKE.ML.DOCUMENT_INTELLIGENCE ON SCHEMA doc_ai_db.doc_ai_schema TO ROLE doc_ai_role;

-- カスタムロールにストリームとタスクを使用して処理パイプラインを作成するために必要な権限を付与
GRANT CREATE STREAM, CREATE TABLE, CREATE TASK, CREATE VIEW ON SCHEMA doc_ai_db.doc_ai_schema TO ROLE doc_ai_role;
GRANT EXECUTE TASK ON ACCOUNT TO ROLE doc_ai_role;

-- おまけ：カスタムロールをSYSADMIN配下に
GRANT ROLE doc_ai_role TO ROLE SYSADMIN;

-- 操作するユーザーに、カスタムロールの権限を付与
GRANT ROLE doc_ai_role TO USER <your_user_name>;

Prepare a Document AI model build

ここでは、実際にDocument AI model buildを作っていきます。Document AI model buildには、モデル、抽出するデータ値の情報、モデルをテストするためにアップロードされたドキュメントが含まれます。

Create a Document AI model build

まず、先程作成したカスタムロールに切り替えます。

Snowsightのメニューで、Document AIを押します。

Select Warehouseを押して、先程作成したウェアハウスを選択します。

右上の+ Buildを押します。

下図の画面が表示されますので、以下の内容を入力してCreateを押します。

• Build name：inspection_reviews
• Location：データベースはDOC_AI_DB、スキーマはDOC_AI_SCHEMA

下図のように表示されれば、Document AI model buildの作成は完了です！

Upload documents to the Document AI model build

チュートリアルの「Prepare a Document AI model build」の「Upload documents to the Document AI model build」1番の項目に書かれているzip fileをクリックし、ダウンロードして解凍します。

実際に一つPDFを開いてみると、装置の検査結果について手書きで書かれたPDFであることがわかります。

ダウンロードして解凍を終えたら、Document AIの画面に戻り、作成したmodel buildのBuild DetailsタブでUpload documentsを押します。

下図の画面が表示されるので、ダウンロードして解凍を行ったPDFをドラッグ＆ドロップでアップロード対象として選択し、右下のUploadを押します。

アップロード後、下図のように表示されていればOKです。

Define data values and review the results

ここでは、「ドキュメントからどのデータを抽出するか」の定義を行います。

Document AIの画面に戻り、作成したmodel buildのBuild DetailsタブでDefine valuesを押します。

下図のような画面となるので、+ Valueを押します。

チュートリアルの内容に沿って、以下の4項目をvalueとして追加していきます。

• inspection_date: What is the inspection date?
• inspection_grade: What is the grade?
• inspector: Who performed the inspection?
• list_of_units: What are all the units?

入力後、入力した各質問に対して正しく値を抽出できているかを確認します。もし間違いがあれば、各値を手動で修正します。

問題なければ、ページ下部のAccept all and review nextを押して、別のPDFも問題ないかを見ていきます。

すべてのドキュメントの確認を終えると、下図のように表示されます。これでアップロードしたドキュメントに対してどのデータを取得するかの定義は完了です！

Publish a Document AI model build

ここでは、ドキュメントをアップロードして取得するデータを定義し終えたDocument AI model buildを公開して、後述するパイプラインで使用できるようにします。

Document AIの画面に戻り、作成したmodel buildのBuild DetailsタブでPublish versionを押します。

下図のように表示されたら、Publishを押します。

Publish後、Build versionが表示され、さらに最下部にこのDocument AI model buildを使用するためのクエリが表示されます。これで公開完了なので、SQL経由で構築したDocument AI model buildを使用できるようになりました。

Create a document processing pipeline

ここでは、先ほど構築したDocument AI model buildと、ストリーム、タスクを用いて新しいドキュメントがアップロードされたときに自動でテーブルデータにロードできるようにパイプラインを構築していきます。

Set up the processing pipeline

ワークシートを開き、順番にクエリを実行していきます。（USEコマンドを先にすべて実行するように、順番をチュートリアルから少しだけ変更しています。）

使用するロール・データベース・スキーマを指定します。

USE ROLE doc_ai_role;
USE DATABASE doc_ai_db;
USE SCHEMA doc_ai_schema;

ドキュメントを保存するための内部ステージを作成します。

CREATE OR REPLACE STAGE my_pdf_stage
  DIRECTORY = (ENABLE = TRUE)
  ENCRYPTION = (TYPE = 'SNOWFLAKE_SSE');

作成した内部ステージに対するストリームを作成します。ステージに対するストリームは、ディレクトリテーブルを有効化しているときに設定可能で、これによりストリームには、ステージに新しいファイルが追加されたときにデータ追加されるようになります。

CREATE STREAM my_pdf_stream ON STAGE my_pdf_stage;

ステージングされたドキュメントファイルを保存するディレクトリテーブルのメタデータを更新します。

ALTER STAGE my_pdf_stage REFRESH;

ドキュメントから抽出したデータを格納するテーブルを作成します。

json_content列を定義しており、このカラムにJSON形式でDocument AIから抽出された情報が含まれるようにしています。

CREATE OR REPLACE TABLE pdf_reviews (
  file_name VARCHAR,
  file_size VARIANT,
  last_modified VARCHAR,
  snowflake_file_url VARCHAR,
  json_content VARCHAR
);

ステージに入ってきた新しいドキュメントを処理するタスクを作成します。

このタスクは、ストリームにデータがある＝新しいファイルが追加されたときに、先ほど構築したDocument AI model buildであるINSPECTION_REVIEWSを用いてデータを抽出しjson_content列にINSERTするということを行っています。

CREATE OR REPLACE TASK load_new_file_data
  WAREHOUSE = doc_ai_wh
  SCHEDULE = '1 minute'
  COMMENT = 'Process new files in the stage and insert data into the pdf_reviews table.'
WHEN SYSTEM$STREAM_HAS_DATA('my_pdf_stream')
AS
INSERT INTO pdf_reviews (
  SELECT
    RELATIVE_PATH AS file_name,
    size AS file_size,
    last_modified,
    file_url AS snowflake_file_url,
    inspection_reviews!PREDICT(GET_PRESIGNED_URL('@my_pdf_stage', RELATIVE_PATH), 1) AS json_content
  FROM my_pdf_stream
  WHERE METADATA$ACTION = 'INSERT'
);

最後に、作成したタスクをRESUME状態に変更します。

ALTER TASK load_new_file_data RESUME;

Upload new documents to an internal stage

実際に作成したステージに新しいPDFをアップロードしてみます。

このチュートリアルの「Create a document processing pipeline」の「Upload new documents to an internal stage」の1番の項目にあるzip fileをダウンロードし、解凍します。

この状態で、Snowsightの画面で先ほど作成したmy_pdf_stageを開き、右上の+ Filesを押します。

ダウンロードして解凍したPDFをすべてドラッグ＆ドロップして、右下のUploadを押します。

View the extracted information

ドキュメントをステージにアップロードしたことで、先程定義したタスクが動作しテーブルにデータがロードされたはずです。

実際に下記のクエリを実行してテーブルを確認してみると、Document AIを用いて抽出したデータがjson_content列に含まれていることがわかります。

SELECT * FROM pdf_reviews;

このままではJSONの形式で分析や活用が難しい状況なので、FLATTEN関数を用いてJSONの一つ一つの要素をカラムに変換したいと思います。

下記のクエリが、FLATTEN関数を用いてJSONの一つ一つの要素をカラムに変換してテーブルとするためのクエリとなります。

CREATE OR REPLACE TABLE doc_ai_db.doc_ai_schema.pdf_reviews_2 AS (
 WITH temp AS (
   SELECT
     RELATIVE_PATH AS file_name,
     size AS file_size,
     last_modified,
     file_url AS snowflake_file_url,
     inspection_reviews!PREDICT(get_presigned_url('@my_pdf_stage', RELATIVE_PATH), 1) AS json_content
   FROM directory(@my_pdf_stage)
 )

 SELECT
   file_name,
   file_size,
   last_modified,
   snowflake_file_url,
   json_content:__documentMetadata.ocrScore::FLOAT AS ocrScore,
   f.value:score::FLOAT AS inspection_date_score,
   f.value:value::STRING AS inspection_date_value,
   g.value:score::FLOAT AS inspection_grade_score,
   g.value:value::STRING AS inspection_grade_value,
   i.value:score::FLOAT AS inspector_score,
   i.value:value::STRING AS inspector_value,
   ARRAY_TO_STRING(ARRAY_AGG(j.value:value::STRING), ', ') AS list_of_units
 FROM temp,
   LATERAL FLATTEN(INPUT => json_content:inspection_date) f,
   LATERAL FLATTEN(INPUT => json_content:inspection_grade) g,
   LATERAL FLATTEN(INPUT => json_content:inspector) i,
   LATERAL FLATTEN(INPUT => json_content:list_of_units) j
 GROUP BY ALL
);

上記のクエリを実行後、作成したテーブルをクエリするとDocument AIを通じて抽出したデータが一つ一つのカラムに変換されていることがわかります。これならば様々な用途に活用ができますね！

SELECT * FROM pdf_reviews_2;

おまけ：お片付け

チュートリアルの内容としてはここまでなのですが、おまけでチュートリアルで作成した各オブジェクトを削除しておきます。

まず、doc_ai_roleを使用する状態でDocument AIの画面を開き、作成したINSPECTION_REVIEWSを削除します。

次に、下記のクエリを実行してチュートリアルで作成した各オブジェクトを削除します。

USE ROLE ACCOUNTADMIN;
DROP DATABASE doc_ai_db;
DROP ROLE doc_ai_role;
DROP WAREHOUSE doc_ai_wh;

これでチュートリアルのお片付けは完了です！

最後に

SnowflakeのDocument AI機能を一通り試すことが出来るチュートリアル「Tutorial: Create a document processing pipeline with Document AI」の内容をまとめてみました。

非常に簡単に出来ることがわかりましたし、ディレクトリテーブルと併用することで新しいファイルがアップロードされたときに抽出処理を行うタスクを作れるので運用面もイメージできる良いチュートリアルでした！

あとは日本語対応ができるといいのですが…今後のアップデートに期待しています！